package cn.tecnova.cd

import cn.tecnova.utils.ConfigHandler
import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{Dataset, SQLContext}
import org.elasticsearch.spark.rdd.EsSpark

/**
  * description:
  * 1214 7508
  **/
object TecnovaTrademark {
  Logger.getLogger("org").setLevel(Level.ERROR)

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
      .setAppName(this.getClass.getSimpleName)
      //      .setMaster("local[*]")
      .set("es.index.auto.create", "true")
      .set("es.nodes", ConfigHandler.esNodes)
      .set("es.port", ConfigHandler.esPort)
      .set("es.nodes.wan.only", "true")

    val sc = new SparkContext(conf)
    val sqLContext = new SQLContext(sc)

    import sqLContext.implicits._
    val rowDF = sqLContext.read.jdbc("jdbc:mysql://www.slave4.com:3306/tec_trademark", "qichacha_trademark", "id", 1, 12147508, 10, ConfigHandler.props2)

    //导入写es相关的包
    import org.elasticsearch.spark.sql._

    val result: Dataset[TecnovaTrademark] = rowDF.map(row => {
      TecnovaTrademark(
        row.getAs[Int]("id"),
        row.getAs[String]("KeyNo"),
        row.getAs[String]("TID"),
        row.getAs[String]("RegNo"),
        row.getAs[String]("Name"),
        row.getAs[Int]("CategoryId"),
        row.getAs[String]("Category"),
        row.getAs[String]("Person"),
        row.getAs[Boolean]("HasImage").toString,
        row.getAs[String]("Flow"),
        row.getAs[String]("ImageUrl"),
        row.getAs[String]("FlowStatus"),
        row.getAs[String]("FlowStatusDesc"),
        row.getAs[String]("AppDate"),
        row.getAs[String]("Status")

      )
    })

    //    result.saveToEs("tecnova_trademark2" + "/company_trademark")
    EsSpark.saveToEs(result.rdd, "tecnova_trademark/company_trademark", Map("es.mapping.id" -> "id"))

    sc.stop()

  }

}

case class TecnovaTrademark(

                             var id: Long,
                             var keyNo: String,
                             var tID: String,
                             var regNo: String,
                             var name: String,
                             var categoryId: Int,
                             var category: String,
                             var person: String,
                             var hasImage: String,
                             var flow: String,
                             var imageurl: String,
                             var flowStatus: String,
                             var flowStatusDesc: String,
                             var appDate: String,
                             var status: String

                           )
